package avicit.bdp.dcs.datasource.rest;

import avicit.bdp.common.utils.database.ColumnInfo;
import avicit.bdp.dcs.datasource.dto.JobDatasource;
import avicit.bdp.dcs.datasource.service.BaseDbService;
import avicit.bdp.dcs.datasource.service.DataSourceUtils;
import avicit.bdp.dcs.datasource.service.DsDatasourceService;
import avicit.platform6.core.rest.msg.ResponseMsg;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import io.swagger.annotations.ApiParam;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;

import java.util.List;
import java.util.Map;

/**
 * @金航数码科技有限责任公司
 * @作者：developer
 * @邮箱：developer@avic-digital.com
 * @创建时间： 2020-06-29 11:55
 * @类说明：数据源接口
 * @修改记录：
 */
@RestController
@RequestMapping("/api/bdp/dcs/datasource/DatasourceRest")
@Api(tags = "数据源接口")
public class DatasourceRest {

    private static final Logger logger = LoggerFactory.getLogger(DatasourceRest.class);

    @Autowired
    private DsDatasourceService dsDatasourceService;
    @Autowired
    private BaseDbService baseDbService;
    @Autowired
    private DataSourceUtils dataSourceUtils;

    @GetMapping("/getDatasourceList")
    @ApiOperation(value = "查询数据源")
    public ResponseMsg<List<Map<String, Object>>> getDatasourceList(@ApiParam(value = "job类型", name = "jobType") @RequestParam(required = false) Integer jobType,
                                                                    @ApiParam(value = "项目空间", name = "projectId") @RequestParam(required = false) String projectId)
            throws Exception{
        ResponseMsg<List<Map<String, Object>>> responseMsg = new ResponseMsg<>();
        responseMsg.setResponseBody(dsDatasourceService.getDatasourceList(jobType, projectId));
        return responseMsg;
    }

    @GetMapping("/getTableListByDatasourceId")
    @ApiOperation(value = "根据数据源ID查询所有表名")
    public ResponseMsg<Map<String, List<String>>> getTableListByDatasourceId(@ApiParam(value = "数据源id", name = "datasourceId", required = true) @RequestParam String datasourceId)
            throws Exception{
        ResponseMsg<Map<String, List<String>>> responseMsg = new ResponseMsg<>();
        responseMsg.setResponseBody(dsDatasourceService.getTableListByDatasourceId(datasourceId));
        return responseMsg;
    }

    @GetMapping("/getPrimaryKeyList")
    @ApiOperation(value = "获取主键列")
    public ResponseMsg<List<ColumnInfo>> getPrimaryKeyList(@ApiParam(value = "任务id", name = "taskId", required = true) @RequestParam String taskId) throws Exception{
        ResponseMsg<List<ColumnInfo>> responseMsg = new ResponseMsg<>();
        responseMsg.setResponseBody(dsDatasourceService.getPrimaryKeyList(taskId));
        return responseMsg;
    }

    @GetMapping("/getDateTimeColumnList")
    @ApiOperation(value = "获取时间类型的列")
    public ResponseMsg<List<ColumnInfo>> getDateTimeColumnList(@ApiParam(value = "任务id", name = "taskId", required = true) @RequestParam String taskId) throws Exception{
        ResponseMsg<List<ColumnInfo>> responseMsg = new ResponseMsg<>();
        responseMsg.setResponseBody(dsDatasourceService.getDateTimeColumnList(taskId));
        return responseMsg;
    }

    @GetMapping("/getTablesByDatasourceId")
    @ApiOperation(value = "根据数据源ID查询所有表名")
    public ResponseMsg<List<String>> getTablesByDatasourceId(@ApiParam(value = "数据源id", name = "datasourceId", required = true) @RequestParam String datasourceId) throws Exception{
        ResponseMsg<List<String>> responseMsg = new ResponseMsg<>();
        responseMsg.setResponseBody(dsDatasourceService.getTablesByDatasourceId(datasourceId));
        return responseMsg;
    }

    @GetMapping("/getColumnList")
    @ApiOperation(value = "根据数据源id和表名获取所有列")
    public ResponseMsg<List<ColumnInfo>> getColumnList(@ApiParam(value = "数据源id", name = "datasourceId", required = true) @RequestParam String datasourceId,
                                                       @ApiParam(value = "表名", name = "tableName", required = true) @RequestParam String tableName) throws Exception{
        ResponseMsg<List<ColumnInfo>> responseMsg = new ResponseMsg<>();
        responseMsg.setResponseBody(baseDbService.getColumnList(datasourceId, tableName));
        return responseMsg;
    }

    @GetMapping("/getDeviceList")
    @ApiOperation(value = "获取设备列表")
    public ResponseMsg<List<String>> getDeviceList(@ApiParam(value = "jobId", name = "jobId", required =
            true) @RequestParam String jobId) throws Exception{
        ResponseMsg<List<String>> responseMsg = new ResponseMsg<>();
        responseMsg.setResponseBody(dsDatasourceService.getDeviceList(jobId));
        return responseMsg;
    }

    @GetMapping("/getKafkaInfoByDataSourceId")
    @ApiOperation(value = "根据数据源ID查询kafka分区和主题")
    public ResponseMsg<Map<String, List<String>>> getKafkaTopicsAndPartionsByDataSourceId(@ApiParam(value =
            "数据源id", name = "datasourceId", required = true) @RequestParam String datasourceId) throws Exception{
        ResponseMsg<Map<String, List<String>>> responseMsg = new ResponseMsg<>();
        responseMsg.setResponseBody(dsDatasourceService.getKafkaTopicsAndPartitionByDataSourceId(datasourceId));
        return responseMsg;
    }

    @GetMapping("/getDatasource")
    @ApiOperation(value = "获取数据源")
    public ResponseMsg<JobDatasource> getDatasource(@RequestParam String datasourceId) throws Exception{
        ResponseMsg<JobDatasource> responseMsg = new ResponseMsg<>();
        responseMsg.setResponseBody(dataSourceUtils.convertDatasource(datasourceId));
        return responseMsg;
    }

    /**
     * 根据数据源ID和主题获取kafka消费者组信息
     * @param datasourceId
     * @param topic
     * @return
     */
    @PostMapping("/getConsumerGroup")
    @ApiOperation(value = "根据数据源ID&主题查询kafka消费者组信息")
    public ResponseMsg<List<String>> getConsumerGroup(
            @ApiParam(value = "数据源id", name = "datasourceId", required = true)
            @RequestParam String datasourceId,
            @ApiParam(value = "主题", name = "topic", required = true)
            @RequestParam String topic) {
        ResponseMsg<List<String>> responseMsg = new ResponseMsg<>();

        responseMsg.setResponseBody(dsDatasourceService.getConsumerGroupByTopic(datasourceId, topic));

        return responseMsg;
    }
}
